package com.hadoop.mapReduce.wordCountStandard;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * Created by xiaoxu on 2018/12/14.
 * mvn 打包命令：mvn clean package -DskipTests
 *
 * Combiner：组合器
 */
public class Combiner {

    /**
     * 1.读取文件
     * Mapper:用到的模板方法模式
     * 数据案例
     * hadoop welcome hadoop
     * hadoop hdfs mapreduce
     * hadoop hdfs
     */
    //-------------------------------------------KEYIN,VALUEIN,KEYOUT,VALUEOUT
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //接收到的每一行数据: hadoop welcome
            String line = value.toString();
            //每一行数据，按照分隔符拆分:hadoop welcome
            String[] words = line.split(" ");

            for (String word : words) {
                //通过上下文，把map处理的结果输出到-->reduce
                context.write(new Text(word), new LongWritable(1));
            }

        }
    }

    /**
     * 2.reduce：计算，归并操作
     * Reducer:用到的模板方法模式
     */
    //-------------------------------------------KEYIN,VALUEIN,KEYOUT,VALUEOUT
    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        // key：某个单词（hadoop）
        // values：集合形式，单词数量的统计[1,1,1,1,1] ==>[2,3]
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0;
            for (LongWritable value : values) {
                sum += value.get();
            }
            context.write(key,new LongWritable(sum));
            //map.put("hadoop",2)
        }
    }

    /**
     * 驱动:封装了MapReduce作业的所有信息
     */
    public static void main(String[] args)throws Exception {
        //创建configuration
        Configuration configuration = new Configuration();
        configuration.set("mapreduce.job.ququename","dev");

        //todo 删除已存在的输出目录
        Path outPutPath = new Path(args[1]);
        FileSystem fileSystem = FileSystem.get(configuration);
        if (fileSystem.exists(outPutPath)){
            fileSystem.delete(outPutPath,true);
            System.out.println("------------file out put is delete------------");
        }

        //创建job
        Job job = Job.getInstance(configuration,"Combiner");
        //设置作业的处理类(反射机制)
        job.setJarByClass(Combiner.class);

        //设置map相关参数
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置reduce相关参数
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        job.setCombinerClass(MyReducer.class);

        //todo 设置combiner处理类 作业输出日志会有combiner关键字
        job.setCombinerClass(MyReducer.class);

        //设置作业处理文件的路径
        FileInputFormat.setInputPaths(job,new Path(args[0]));

        //设置作业输出路径
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        //等待作业执行完毕
        int status = job.waitForCompletion(true)?0:1;
        //退出程序
        System.exit(status);
    }
}
